15 人在环路

Agent有时候会做一些不该做的事——比如误删数据、发错邮件、执行危险的SQL。你不能完全信任AI的判断，尤其是在涉及重要操作的时候。

人在环路（Human-in-the-Loop，简称HITL）就是来解决这个问题的：在敏感操作执行前暂停，等人工确认后再继续。

打个比方：Agent就像一个实习生，能力不错但经验不足。重要的事情不能让他自己做主，需要你签字确认才行。

一、基本用法

LangChain提供了内置的HumanInTheLoopMiddleware中间件，配置很简单：

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver


def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    # 实际的邮件发送逻辑
    return f"邮件已发送给 {to}"


def delete_record(table: str, record_id: str) -> str:
    """删除数据库记录"""
    return f"已删除 {table} 表中 ID 为 {record_id} 的记录"


def query_data(sql: str) -> str:
    """查询数据"""
    return "查询结果: ..."


agent = create_agent(
    model="deepseek-v4-flash",
    tools=[send_email, delete_record, query_data],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": True,    # 发邮件前需要确认
                "delete_record": True, # 删除操作前需要确认
                "query_data": False,   # 查询是安全操作，不需要确认
            },
        ),
    ],
    # 必须配置checkpointer来保存状态
    checkpointer=InMemorySaver(),
)

interrupt_on的配置规则：

值	含义
True	需要确认，允许所有决策（批准、编辑、拒绝、回复）
False	不需要确认，自动执行
{"allowed_decisions": ["approve", "reject"]}	需要确认，但只允许特定决策

二、四种决策类型

当中断发生时，你可以做出四种不同的决策：

2.1 approve - 批准

原样执行Agent想要做的操作：

from langgraph.types import Command

config = {"configurable": {"thread_id": "conversation_1"}}

# 运行Agent，遇到中断会暂停
result = agent.invoke(
    {"messages": [{"role": "user", "content": "给张三发一封会议邀请邮件"}]},
    config=config,
    version="v2",
)

# 查看需要审批的操作
print(result.interrupts)
# 包含：send_email, args: {to: "zhangsan@example.com", subject: "会议邀请", ...}

# 批准执行
agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
    version="v2",
)

2.2 edit - 修改后执行

修改参数再执行。比如你想改一下邮件的收件人：

agent.invoke(
    Command(resume={
        "decisions": [{
            "type": "edit",
            "edited_action": {
                "name": "send_email",
                "args": {
                    "to": "lisi@example.com",  # 改成李四
                    "subject": "会议邀请",
                    "body": "请参加明天的会议",
                },
            },
        }]
    }),
    config=config,
    version="v2",
)

注意：编辑参数时要保守一点，改动太大的话模型可能会重新评估策略，导致意想不到的行为。

2.3 reject - 拒绝

拒绝执行，并告诉Agent为什么：

agent.invoke(
    Command(resume={
        "decisions": [{
            "type": "reject",
            "message": "不能删除这条记录，这是重要的客户数据。请改为归档处理。",
        }]
    }),
    config=config,
    version="v2",
)

拒绝时的message会作为反馈加入对话历史，Agent会看到这个反馈并调整后续行为。

2.4 respond - 回复

跳过工具执行，把人的回复直接作为工具结果。适合"询问用户"类型的工具：

# 假设Agent调用了一个 ask_user 工具来询问用户偏好
agent.invoke(
    Command(resume={
        "decisions": [{
            "type": "respond",
            "message": "我喜欢蓝色。",
        }]
    }),
    config=config,
    version="v2",
)

这种方式下，工具本身不会被执行，人的回复直接成为工具的返回值。

三、完整流程示例

来看一个完整的例子，模拟一个数据库管理Agent：

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command


# 定义工具
def execute_sql(query: str) -> str:
    """执行SQL语句"""
    return f"执行结果: {query}"


def backup_table(table_name: str) -> str:
    """备份表"""
    return f"表 {table_name} 已备份"


# 创建Agent
agent = create_agent(
    model="deepseek-v4-flash",
    tools=[execute_sql, backup_table],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "execute_sql": {
                    "allowed_decisions": ["approve", "reject"],
                    "description": "SQL执行需要DBA审批",
                },
                "backup_table": False,  # 备份是安全操作
            },
        ),
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "db_admin"}}

# 第一步：用户请求删除数据
result = agent.invoke(
    {"messages": [{"role": "user", "content": "删除30天前的日志记录"}]},
    config=config,
    version="v2",
)

# Agent会生成SQL并请求执行，此时被中断
print("中断信息:", result.interrupts)

# 第二步：DBA审核后拒绝，并给出建议
agent.invoke(
    Command(resume={
        "decisions": [{
            "type": "reject",
            "message": "直接删除太危险了。请先导出备份，再执行删除。",
        }]
    }),
    config=config,
    version="v2",
)

# Agent会看到拒绝反馈，调整策略：先备份再删除
# 可能会调用 backup_table（不需要审批）然后再请求 execute_sql

四、流式输出配合HITL

在流式模式下也可以处理中断：

config = {"configurable": {"thread_id": "stream_demo"}}

# 流式运行，直到遇到中断
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "删除旧数据"}]},
    config=config,
    stream_mode=["updates", "messages"],
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]
        if token.content:
            print(token.content, end="", flush=True)
    elif chunk["type"] == "updates":
        if "__interrupt__" in chunk["data"]:
            print(f"\n\n需要审批: {chunk['data']['__interrupt__']}")

# 人工审批后，继续流式输出
for chunk in agent.stream(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
    stream_mode=["updates", "messages"],
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]
        if token.content:
            print(token.content, end="", flush=True)

五、多个操作同时审批

有时候Agent一次会调用多个需要审批的工具。这时候你需要按顺序为每个操作提供决策：

# Agent同时要发邮件和删除记录
result = agent.invoke(
    {"messages": [{"role": "user", "content": "给张三发确认邮件，然后删除临时数据"}]},
    config=config,
    version="v2",
)

# 为每个操作分别做决策
agent.invoke(
    Command(resume={
        "decisions": [
            {"type": "approve"},  # 批准发邮件
            {
                "type": "edit",   # 修改删除条件
                "edited_action": {
                    "name": "delete_record",
                    "args": {"table": "temp_data", "record_id": "batch_001"},
                },
            },
        ]
    }),
    config=config,
    version="v2",
)

决策的顺序必须和中断请求中操作的顺序一致。

六、工作原理

HITL中间件的执行流程：

1. Agent调用模型生成响应
2. 中间件在after_model钩子中检查响应中的工具调用
3. 如果有工具调用匹配了interrupt_on中的配置，构建中断请求
4. 发出interrupt，保存当前图状态（通过checkpointer）
5. 等待人工决策
6. 根据决策类型处理：
   • approve：执行原工具调用
   • edit：用修改后的参数执行
   • reject：生成拒绝的ToolMessage
   • respond：把人的回复作为ToolMessage
7. 继续执行

关键点：必须配置checkpointer。生产环境用持久化的checkpointer（比如AsyncPostgresSaver），开发测试用InMemorySaver。同时每次调用都要传config包含thread_id，这样才能正确关联暂停和恢复的对话。

七、总结

人在环路让Agent在敏感操作前暂停等待人工确认：

• 四种决策：批准、修改、拒绝、回复
• 灵活配置：可以按工具配置是否需要审批、允许哪些决策
• 安全可靠：通过checkpointer持久化状态，支持暂停和恢复
• 流式支持：在流式模式下也能处理中断

HITL是把Agent从"玩具"变成"生产工具"的关键机制。有了它，你就可以放心地让Agent处理重要任务，因为你知道关键操作会经过人工审核。

在下一篇文章中，我们将学习Runtime运行时信息，了解如何在Agent执行过程中获取和使用运行时上下文。